"""
Case Type   : 基础功能
Case Name   : 同步一级分区表基本数据
Create At   : 2022/11/08
Owner       : opentestcase026
Description :
    1.两个集群均创建表
      1.1 创建range一级分区表
      1.2 创建list一级分区表
      1.3 创建hash一级分区表
      1.4 创建interval一级分区表
    2.创建发布端
    3.创建订阅端
    4.发布端插入基础数据
    5.查询订阅端表数据
      5.1 查询range一级分区表数据
      5.2 查询list一级分区表数据
      5.3 查询hash一级分区表数据
      5.4 查询interval一级分区表数据
    6. 删除订阅
    7. 删除发布
    8. 清理环境
Expect      :
    1.成功
    2.成功
    3.成功
    4.成功
    5.数据同步成功
      5.1 range一级分区表数据同步成功
      5.2 list一级分区表数据同步成功
      5.3 hash一级分区表数据同步成功
      5.4 interval一级分区表数据同步成功
    6.成功
    7.成功
    8.成功
History     :
"""
import unittest
import os
from yat.test import macro
from yat.test import Node
from testcase.utils.Logger import Logger
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Common import Common
from testcase.utils.Constant import Constant

Primary_SH = CommonSH('PrimaryDbUser')


@unittest.skipIf(3 != Primary_SH.get_node_num(), '非1+2环境不执行')
class Pubsubclass(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        self.log.info("-----------this is setup-----------")
        self.log.info(f"-----{os.path.basename(__file__)[:-3]} start-----")
        self.pri_userdb_pub = Node(node='PrimaryDbUser')
        self.pri_userdb_sub = Node(node='remote1_PrimaryDbUser')
        self.constant = Constant()
        self.commsh_pub = CommonSH('PrimaryDbUser')
        self.commsh_sub = CommonSH('remote1_PrimaryDbUser')
        self.com_pub = Common()
        self.tb_name1 = 'tb_pubsub_case235_01'
        self.tb_name2 = 'tb_pubsub_case235_02'
        self.tb_name3 = 'tb_pubsub_case235_03'
        self.tb_name4 = 'tb_pubsub_case235_04'
        self.subname = "sub_case235"
        self.pubname = "pub_case235"
        self.port = str(int(self.pri_userdb_pub.db_port) + 1)
        self.wal_level = self.com_pub.show_param("wal_level")
        self.user_param_pub = f'-U {self.pri_userdb_pub.db_user} ' \
            f'-W {self.pri_userdb_pub.db_password}'
        self.user_param_sub = f'-U {self.pri_userdb_sub.db_user} ' \
            f'-W {self.pri_userdb_sub.db_password}'
        text = '----修改pg_hba expect:成功----'
        self.log.info(text)
        self.log.info('----发布端状态----')
        status = self.commsh_pub.get_db_cluster_status('detail')
        self.log.info(status)
        self.assertEqual(status.count('Normal'), 4, '执行失败' + text)
        self.log.info('----订阅端状态----')
        status = \
            self.commsh_sub.get_db_cluster_status(
                'detail', env_path=macro.DB_ENV_PATH_REMOTE1)
        self.log.info(status)
        self.assertEqual(status.count('Normal'), 4, '执行失败' + text)
        guc_res = self.commsh_pub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, '',
            f'host    replication  {self.pri_userdb_sub.db_user} '
            f'{self.pri_userdb_sub.db_host}/32 sha256')
        self.log.info(guc_res)
        self.assertTrue(guc_res, '执行失败:' + text)
        if 'logical' != self.wal_level:
            result = self.commsh_pub.execute_gsguc(
                'set', self.constant.GSGUC_SUCCESS_MSG, 'wal_level=logical')
            self.assertTrue(result, '执行失败:' + text)
            result = self.commsh_pub.restart_db_cluster(True)
            flg = self.constant.START_SUCCESS_MSG in result \
                  or 'Degrade' in result
            self.assertTrue(flg, '执行失败:' + text)
        guc_res = self.commsh_sub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, macro.DB_INSTANCE_PATH_REMOTE1,
            f'host    replication  {self.pri_userdb_pub.db_user} '
            f'{self.pri_userdb_pub.db_host}/32 sha256',
            macro.DB_ENV_PATH_REMOTE1)
        self.log.info(guc_res)
        self.assertTrue(guc_res, '执行失败:' + text)

    def test_pubsub(self):
        text = '--step1:两个集群均创建表 expect:成功--'
        self.log.info(text)
        text = '--step1.1:两个集群均创建range一级分区表 expect:成功--'
        self.log.info(text)
        sql = f'''create table {self.tb_name1}
                 (
                    id int primary key constraint id_nn not null,
                    use_filename varchar2(20),
                    filename varchar2(255),
                    text varchar2(2000)
                 )
                  partition by range(id)
                 (
                    partition p1 values less than(30),
                    partition p2 values less than(60),
                    partition p3 values less than(90),
                    partition p4 values less than(maxvalue)
                 );'''
        result = self.commsh_pub.execut_db_sql(
            sql, sql_type=self.user_param_pub)
        self.log.info(result)
        self.assertEqual(result.count(self.constant.TABLE_CREATE_SUCCESS),
                         2, '执行失败:' + text)
        result = self.commsh_sub.execut_db_sql(sql, self.user_param_sub, None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertEqual(result.count(self.constant.TABLE_CREATE_SUCCESS),
                         2, '执行失败:' + text)
        text = '--step1.2:两个集群均创建list一级分区表 expect:成功--'
        self.log.info(text)
        sql = f'''create table {self.tb_name2}
                 (
                    c_id integer primary key,
                    c_name varchar(16) default 'omm' NOT NULL,
                    c_class varchar(20) not null
                 )
                   partition by LIST(c_id)
                 (
                     partition p1 values (1,2,3),
                     partition p2 values (4,5,6),
                     partition p3 values (7,8,9),
                     partition p4 values (10,11,12),
                     partition p5 values (13,14,15),
                     partition p6 values (16,17,18)
                 );'''
        result = self.commsh_pub.execut_db_sql(
            sql, sql_type=self.user_param_pub)
        self.log.info(result)
        self.assertEqual(result.count(self.constant.TABLE_CREATE_SUCCESS),
                         2, '执行失败:' + text)
        result = self.commsh_sub.execut_db_sql(sql, self.user_param_sub, None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertEqual(result.count(self.constant.TABLE_CREATE_SUCCESS),
                         2, '执行失败:' + text)
        text = '--step1.3:两个集群均创建hash一级分区表 expect:成功--'
        self.log.info(text)
        sql = f'''create table {self.tb_name3}
                 (
                    c_id integer primary key,
                    c_name varchar(16) default 'omm',
                    c_class varchar(20) not null
                 )
                  partition by hash(c_id)
                 (
                    partition p1,
                    partition p2,
                    partition p3,
                    partition p4,
                    partition p5,
                    partition p6
                 );'''
        result = self.commsh_pub.execut_db_sql(
            sql, sql_type=self.user_param_pub)
        self.log.info(result)
        self.assertEqual(result.count(self.constant.TABLE_CREATE_SUCCESS),
                         2, '执行失败:' + text)
        result = self.commsh_sub.execut_db_sql(sql, self.user_param_sub, None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertEqual(result.count(self.constant.TABLE_CREATE_SUCCESS),
                         2, '执行失败:' + text)
        text = '--step1.4:两个集群均创建interval一级分区表 expect:成功--'
        self.log.info(text)
        sql = f'''create table {self.tb_name4}
                 (
                   c_id integer ,
                   c_date DATE primary key,
                   c_info varchar(20) not null
                 )
                 partition by range(c_date)
                 interval('2 day')
                 (
                 partition p1 values less than ('2021-06-01 00:00:00'),
                 partition p2 values less than ('2021-06-03 00:00:00'),
                 partition p3 values less than ('2021-06-05 00:00:00'),
                 partition p4 values less than ('2021-06-07 00:00:00'),
                 partition p5 values less than ('2021-06-09 00:00:00')
                 );'''
        result = self.commsh_pub.execut_db_sql(
            sql, sql_type=self.user_param_pub)
        self.log.info(result)
        self.assertEqual(result.count(self.constant.TABLE_CREATE_SUCCESS),
                         2, '执行失败:' + text)
        result = self.commsh_sub.execut_db_sql(sql, self.user_param_sub, None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertEqual(result.count(self.constant.TABLE_CREATE_SUCCESS),
                         2, '执行失败:' + text)

        text = '--step2:创建发布端;expect:成功--'
        self.log.info(text)
        sql = f"CREATE PUBLICATION {self.pubname} " \
              f"FOR all TABLEs;"
        result = self.commsh_pub.execut_db_sql(sql,
                                               sql_type=self.user_param_pub)
        self.log.info(result)
        self.assertIn(self.constant.create_pub_succ_msg, result,
                      '执行失败:' + text)
        self.assertNotIn(self.constant.SQL_WRONG_MSG[1], result,
                         '执行失败:' + text)

        text = '--step3:创建订阅端;expect:成功--'
        self.log.info(text)
        result = self.commsh_sub.execute_generate(
            macro.COMMON_PASSWD, env_path=macro.DB_ENV_PATH_REMOTE1)
        self.assertIn('', result, '执行失败:' + text)
        sql = f"CREATE SUBSCRIPTION {self.subname} CONNECTION " \
            f"'host={self.pri_userdb_pub.db_host} " \
            f"port={self.port} " \
            f"user={self.pri_userdb_pub.db_user} " \
            f"dbname={self.pri_userdb_pub.db_name} " \
            f"password={self.pri_userdb_pub.ssh_password}' " \
            f"PUBLICATION {self.pubname} with (copy_data=false);"
        result = self.commsh_sub.execut_db_sql(sql, self.user_param_sub, None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertIn(self.constant.create_sub_succ_msg,
                      result, '执行失败:' + text)

        text = '--step4:发布端插入数据 expect:成功--'
        self.log.info(text)
        sql = f"insert into {self.tb_name1} values(1, 'first', '%一', '')," \
              f"(60, 'first', '%二', ''),(90, 'first', '%三', ''); " \
              f"select pg_sleep(5.5);" \
              f"insert into {self.tb_name2} values(1, 'first', '%一')," \
              f"(9, 'first', '%二'),(12, 'first', '%三'); " \
              f"select pg_sleep(5.5);" \
              f"insert into {self.tb_name3} values(1, 'first', '%一')," \
              f"(9, 'first', '%二'),(12, 'first', '%三'); " \
              f"select pg_sleep(5.5);" \
              f"insert into {self.tb_name4} values(1, '2021-05-30 " \
              f"00:00:00', '1-1')," \
              f"(5, '2021-06-03 00:00:00', '1-3');" \
              f"select pg_sleep(5.5);"
        result = self.commsh_pub.execut_db_sql(sql,
                                               sql_type=self.user_param_pub)
        self.log.info(result)
        self.assertIn(self.constant.INSERT_SUCCESS_MSG, result, '执行失败' +
                      text)

        text = '--step5:查询订阅端表数据 expect:成功--'
        self.log.info(text)
        text = '--step5.1:查询订阅端range一级分区表数据 expect:成功--'
        self.log.info(text)
        sql = f"select * from {self.tb_name1};"
        result = self.commsh_sub.execut_db_sql(sql, self.user_param_sub, None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertEqual(result.count('3 row'), 1, '执行失败' + text)
        self.assertIn(' 90 | first        | %三',
                      result, '执行失败:' + text)
        text = '--step5.2:查询订阅端list一级分区表数据 expect:成功--'
        self.log.info(text)
        sql = f"select * from {self.tb_name2};"
        result = self.commsh_sub.execut_db_sql(sql, self.user_param_sub, None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertEqual(result.count('3 row'), 1, '执行失败' + text)
        self.assertIn('12 | first  | %三',
                      result, '执行失败:' + text)
        text = '--step5.3:查询订阅端hash一级分区表数据 expect:成功--'
        self.log.info(text)
        sql = f"select * from {self.tb_name3};"
        result = self.commsh_sub.execut_db_sql(sql, self.user_param_sub, None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertEqual(result.count('3 row'), 1, '执行失败' + text)
        self.assertIn('12 | first  | %三',
                      result, '执行失败:' + text)
        text = '--step5.4:查询订阅端interval一级分区表数据 expect:成功--'
        self.log.info(text)
        sql = f"select * from {self.tb_name4};"
        result = self.commsh_sub.execut_db_sql(sql, self.user_param_sub, None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertEqual(result.count('2 row'), 1, '执行失败' + text)
        self.assertIn(' 1 | 2021-05-30 00:00:00 | 1-1',
                      result, '执行失败:' + text)

    def tearDown(self):
        self.log.info('------------this is tearDown-------------')
        text = '--step6:删除订阅 expect:成功--'
        self.log.info(text)
        sql = f"DROP SUBSCRIPTION  {self.subname};"
        drop_sub_result = self.commsh_sub.execut_db_sql(
            sql, self.user_param_sub, None, macro.DB_ENV_PATH_REMOTE1)
        self.log.info(drop_sub_result)

        text = '--step7:删除发布 expect:成功--'
        self.log.info(text)
        sql = f"DROP PUBLICATION {self.pubname};"
        drop_pub_result = self.commsh_pub.execut_db_sql(
            sql, sql_type=self.user_param_pub)
        self.log.info(drop_pub_result)

        text = '--step8:清理环境 expect:成功--'
        self.log.info(text)
        sql = f"DROP table {self.tb_name1},{self.tb_name2}," \
              f"{self.tb_name3},{self.tb_name4};"
        result = self.commsh_sub.execut_db_sql(sql, self.user_param_sub, None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        result = self.commsh_pub.execut_db_sql(sql, 
                                               sql_type=self.user_param_pub)
        self.log.info(result)
        result1 = self.commsh_pub.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG,
            f'wal_level={self.wal_level}')
        self.log.info(result1)
        guc_res1 = self.commsh_pub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, '',
            f'host    replication  {self.pri_userdb_sub.db_user} '
            f'{self.pri_userdb_sub.db_host}/32')
        self.log.info(guc_res1)
        guc_res = self.commsh_sub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, macro.DB_INSTANCE_PATH_REMOTE1,
            f'host    replication  {self.pri_userdb_pub.db_user} '
            f'{self.pri_userdb_pub.db_host}/32',
            macro.DB_ENV_PATH_REMOTE1)
        self.log.info(guc_res)
        self.assertTrue(result1, '执行失败:' + text)
        self.assertTrue(guc_res, '执行失败:' + text)
        self.assertTrue(guc_res1, '执行失败:' + text)
        self.assertIn(self.constant.drop_pub_succ_msg, drop_pub_result,
                      '执行失败' + text)
        self.assertIn(self.constant.drop_sub_succ_msg, drop_sub_result,
                      '执行失败' + text)
        result = self.commsh_pub.restart_db_cluster(True)
        flg = self.constant.START_SUCCESS_MSG in result \
              or 'Degrade' in result
        self.assertTrue(flg, '执行失败:' + text)
        self.log.info(f"-----{os.path.basename(__file__)[:-3]} end-----")
